package com.itheima.one2many广播模式与集群模式;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 广播模式  consumer.setMessageModel(MessageModel.BROADCASTING);
 * 如果生产者先发送消息,后启动消费者,消息只能被消费一次
 * 如果 多个广播模式消费者 先启动, 后启动消费者 ,才有广播效果
 *
 * 结论:
 *      先启动消费者 再启动发送者才有广播 效果
 */

/**
 * 集群模式 (默认)   consumer.setMessageModel(MessageModel.CLUSTERING);
 *
 * 集群消息每条消息都是集群内的消费者共同消费且不会重复消费
 */

/**
 * https://itzones.cn/2019/07/03/RocketMQ%E6%B6%88%E8%B4%B9%E6%A8%A1%E5%BC%8F/
 */
//消费者，接收消息
public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.创建一个接收消息的对象Consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        System.out.println(consumer.getInstanceName());


        //2.设定接收的命名服务器地址
        consumer.setNamesrvAddr("localhost:9876");
        //3.设置接收消息对应的topic,对应的sub标签为任意*
        consumer.subscribe("topic1","*");

        //设置当前消费者的消费模式（默认模式：负载均衡）
//        consumer.setMessageModel(MessageModel.CLUSTERING);
        //设置当前消费者的消费模式为广播模式：所有客户端接收的消息都是一样的
//        consumer.setMessageModel(MessageModel.BROADCASTING);

        //3.开启监听，用于接收消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //遍历消息
                for(MessageExt msg : list){
//                    System.out.println("收到消息："+msg);
                    System.out.println("Consumer1 ===== "+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //4.启动接收消息的服务
        consumer.start();
        System.out.println("Consumer1接收消息服务已开启运行");
    }
}
